package com.rds.benchmark.command;

import com.rds.benchmark.config.ArgvsConfig;
import com.rds.benchmark.connector.Connector;
import com.rds.benchmark.operator.TypeOperator;
import com.rds.benchmark.thread.StableRunThread;
import com.rds.benchmark.thread.StateThread;

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.CyclicBarrier;

public class StableOperators extends CommandOperator {

    private Class<? extends TypeOperator> operatorClazz;
    private String type;

    public StableOperators(Class<? extends TypeOperator> operatorClazz, String type) {
        this.operatorClazz = operatorClazz;
        this.type = type;
    }

    public void runOperator() throws Exception {
        CyclicBarrier startBarrier = Connector.startBarrier;
        int max_threads = ArgvsConfig.threads;

        init_pxx();

        final TypeOperator[] operators = new TypeOperator[max_threads];
        for (int i = 0; i < max_threads; i++) {
            operators[i] = operatorClazz.getDeclaredConstructor().newInstance();
        }

        Thread[] threads = new Thread[max_threads];

        startBarrier.reset();

        CountDownLatch latch = new CountDownLatch(max_threads);

        for (int i = 0; i < max_threads; i++) {
            StableRunThread stableRunThread = new StableRunThread(Integer.toString(i));
            stableRunThread.initParameter(operators,latch);

            threads[i] = stableRunThread;
            threads[i].start();
        }

        StateThread state = new StateThread();
        state.initParameter(operators,type,latch,"Stable");

        state.start();
        state.join();
    }
}
